package com.shujia.streaming

import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Code03UpdateState {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local[2]").setAppName("wordCountStreaming"))

    sc.setCheckpointDir("spark_code/data/stream/checkpoint")

    val streamingContext = new StreamingContext(sc, Durations.seconds(5))
    val stream: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 8888)

    stream
      .flatMap(_.split(","))
      .map(x => (x, 1))
      // updateStateByKey直接执行会报错： The checkpoint directory has not been set
      // 需要设置checkpoint,该方式用于保证应用程序的容错性
      //  因为有状态更新算子需要保存历史数据，那么当程序宕机，历史数据丢失之后计算结果不准确
      .updateStateByKey(
        //
        (sep: Seq[Int], opt: Option[Int]) => {
          // 通过sep.sum可以对当前批次中所有的Value数据进行求和
          var currentNum: Int = sep.sum
          opt match {
            // 如果匹配到Some表示之前批次中已经通过计算得到该Key的Value值
            case Some(v) => {
              // 对之前批次总和和当前批次总和进行累加
              currentNum += v
              Some(currentNum)
            }
            // 如果为None表示之前批次每有出现该Key值
            case None => Some(currentNum)
          }

        }
      ).print()
    // 启动任务
    streamingContext.start()
    streamingContext.awaitTermination()

  }
}
